package com.cssl.utils;

import com.cssl.bean.TransientSink;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;

/**
 * @Author: chen
 * @Date: 2021/11/17 20:18
 * @Desc:
 */
public class ClickHouseUtil {
    /**
     * JdbcSink.sink()
     * 参数 1: sql语句
     * 参数 2: 给占位符赋值
     * 参数 3: 执行选项 批大小 批次间隔啥的
     * 参数 4: 连接参数 driver url啥的
     */
    public static <T> SinkFunction<T> getSink(String sql) {
        Properties properties = CommonUtils.getProperties();
        String driver = properties.getProperty("clickhouse.driver");
        String url = properties.getProperty("clickhouse.url");
        return JdbcSink.<T>sink(
                sql,
                new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                        try {
                            //获取所有的属性信息
                            Field[] fields = t.getClass().getDeclaredFields();
                            int offset = 0;
                            for (int i = 0; i < fields.length; i++) {

                                //获取字段
                                Field field = fields[i];

                                //设置私有属性可访问
                                field.setAccessible(true);

                                //获取字段上的注解
                                TransientSink annotation = field.getAnnotation(TransientSink.class);
                                if (annotation != null) {
                                    offset++;
                                    continue;
                                }

                                //获取值
                                Object value = field.get(t);

                                //给预编译sql对象赋值
                                preparedStatement.setObject(i + 1 - offset, value);
                            }
                        } catch (IllegalAccessException e) {
                            e.printStackTrace();
                        }
                    }
                },
                new JdbcExecutionOptions.Builder()
                        .withBatchSize(5)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(driver)
                        .withUrl(url)
                        .build());
    }
}
